package com.test.study.ToolsUtils.strom.trident;

import java.util.HashMap;
import java.util.Map.Entry;

import org.apache.storm.trident.operation.BaseFunction;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.tuple.TridentTuple;

/**
 * 汇总所有的数据
 * 
 * @author huyo@SuppressWarnings("serial") ng
 *
 */
public class PrintfFunction extends BaseFunction {

	/**
	 * 
	 */
	private static final long serialVersionUID = 511228477804834366L;

	HashMap<String, Integer> count = new HashMap<String, Integer>();

	@SuppressWarnings("unchecked")
	@Override
	public void execute(TridentTuple tuple, TridentCollector collector) {
		HashMap<String, Integer> hashMap = (HashMap<String, Integer>) (tuple.getValue(0));
		for (Entry<String, Integer> entry : hashMap.entrySet()) {
			String key = entry.getKey();
			Integer value = entry.getValue();
			System.out.println(key + "\t" + value);
			Integer integer = count.get(key);
			if(integer==null) {
				integer=0;
			}
			count.put(key, value + integer);
		}

		System.out.println("================printf final result==============");
		for (Entry<String, Integer> entry : count.entrySet()) {
			String key = entry.getKey();
			Integer value = entry.getValue();
			System.out.println("key:"+key + "\tvalue:" + value);
		}

	}

}
